import os
from pyflink.table import EnvironmentSettings, BatchTableEnvironment

os.environ['SUBMIT_ARGS'] = 'remote -m 10.3.69.162:8081'

env_settings = EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build()
t_env = BatchTableEnvironment.create(environment_settings=env_settings)

source_csv = os.path.join(os.path.abspath(os.path.dirname(__file__)), '../CSV/sort-output.csv')
sink_csv = os.path.join(os.path.abspath(os.path.dirname(__file__)), '../CSV/flink-output.csv')

t_env.execute_sql(f"""
    CREATE TABLE source (
        record_time STRING,     
        source_ip STRING,
        destination_ip STRING,
        protocol STRING,
        destination_port STRING,
        uplink_length INT,
        downlink_length INT
    ) WITH (
        'connector' = 'filesystem',
        'path' = '{source_csv}',
        'format' = 'csv'
    )
""")

t_env.execute_sql(f"""
    CREATE TABLE sink (
        record_time STRING,     
        uplink_length INT,
        downlink_length INT
    ) WITH (
        'connector' = 'filesystem',
        'path' = '{sink_csv}',
        'format' = 'csv'
    )
""")

t_env.execute_sql("""
    INSERT INTO sink
    SELECT record_time,uplink_length,downlink_length         
    FROM source
""")